#include "sys.h"
#include "net.h"
#include "log.h"
#include "process_monitor.h"
#include "thread_pool.h"
#include "platform.h"
#include "cJSON.h"
#include "dial.h"

#define AUTH_JSON_FILE "auth.json"

extern void* specify_dial_task(void* arg);

extern void log_upload_report(char *url, char *probe_id, char *mac);

static time_t _Check_update_tm = 0;

#define _TEST_DATA_ 0

#if _TEST_DATA_
static cJSON* __Test_dial_auth_data()
{
    char* buf = "{\"probeId\":\"1000008\",\"token\":\"C71544C2605146F382EE99FB3E5ADE8C\",\"status\":\"1\"}";
    return cJSON_Parse(buf);
}
static cJSON* __Test_dial_req_data()
{
    char* buf = "{\"dialTaskList\":[{\"dialTaskId\":\"150227934534931\",\"version\":\"1.0\",\"dialTactics\":{\"pingTactics\":[{\"host\":\"10.0.0.210\",\"size\":32,\"port\":18087,\"times\":5,\"timeout\":3,\"businessTag\":\"5\",\"type\":\"tcp\",\"kafkaTopic\":\"pingTopic\"},{\"host\":\"10.0.0.210\",\"size\":32,\"port\":0,\"times\":5,\"timeout\":3,\"businessTag\":\"6\",\"type\":\"icmp\",\"kafkaTopic\":\"pingTopic\"}],\"udpUpTactics\":[],\"httpTactics\":[{\"httpUrlVal\":\"https://hao.qq.com\",\"port\":0,\"timeout\":10,\"businessTag\":\"1\",\"kafkaTopic\":\"httpTopic\"},{\"httpUrlVal\":\"http://www.qq.com\",\"port\":80,\"timeout\":10,\"businessTag\":\"2\",\"kafkaTopic\":\"httpTopic\"}],\"tcpTactics\":[{\"tcpUrlVal\":\"192.168.1.1\",\"port\":45,\"timeout\":2,\"businessTag\":\"3\",\"kafkaTopic\":\"tcpTopic\"},{\"tcpUrlVal\":\"192.168.1.2\",\"port\":46,\"timeout\":2,\"businessTag\":\"4\",\"kafkaTopic\":\"tcpTopic\"}],\"httpDownloadTactics\":[{\"urls\":\"\",\"downTime\":15,\"ignoreTime\":5,\"threads\":4,\"businessTag\":\"11\",\"kafkaTopic\":\"speedTopic\"}],\"tracerouteTactics\":[{\"toAddress\":\"180.97.33.108\",\"size\":32,\"maxNum\":20,\"type\":\"udp\",\"timeout\":3,\"businessTag\":\"7\",\"kafkaTopic\":\"tracerouteTopic\"},{\"toAddress\":\"180.97.33.108\",\"size\":32,\"maxNum\":20,\"type\":\"icmp\",\"timeout\":3,\"businessTag\":\"8\",\"kafkaTopic\":\"tracerouteTopic\"}],\"httpUploadTactics\":[{\"url\":\"\",\"threads\":4,\"upTime\":15,\"ignoreTime\":5,\"businessTag\":\"11\",\"kafkaTopic\":\"speedTopic\"}],\"dnsTactics\":[{\"ip\":\"123.1.2.3\",\"port\":53,\"domainName\":\"www.wangding.com\",\"type\":\"udp\",\"businessTag\":\"9\",\"kafkaTopic\":\"dnsTopic\",\"timeout\":3},{\"ip\":\"123.2.3.5\",\"port\":53,\"domainName\":\"www.wangding2.com\",\"type\":\"tcp\",\"businessTag\":\"10\",\"kafkaTopic\":\"dnsTopic\",\"timeout\":3}]},\"startTime\":1483203661,\"endTime\":1510246861,\"cycleTime\":18000,\"collectTime\":5,\"collectType\":1,\"kafkaUrl\":\"10.0.0.210:9092\"},{\"dialTaskId\":\"1504030622457\",\"version\":\"1.2\",\"dialTactics\":{\"pingTactics\":[],\"udpUpTactics\":[],\"httpTactics\":[{\"httpUrlVal\":\"http://maven.outofmemory.cn/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022003\",\"kafkaTopic\":\"httpTopic\"},{\"httpUrlVal\":\"http://maven.outofmemory.cn/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022003\",\"kafkaTopic\":\"httpTopic\"},{\"httpUrlVal\":\"http://maven.outofmemory.cn/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022003\",\"kafkaTopic\":\"httpTopic\"}],\"tcpTactics\":[],\"httpDownloadTactics\":[],\"tracerouteTactics\":[],\"httpUploadTactics\":[],\"dnsTactics\":[]},\"startTime\":1504159320,\"endTime\":1506060120,\"cycleTime\":60,\"collectTime\":30,\"collectType\":0,\"kafkaUrl\":\"10.0.0.210:9092\"},{\"dialTaskId\":\"1503285489320\",\"version\":\"2.2\",\"dialTactics\":{\"pingTactics\":[],\"udpUpTactics\":[],\"httpTactics\":[{\"httpUrlVal\":\"https://www.baidu.com/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022002\",\"kafkaTopic\":\"httpTopic\"}],\"tcpTactics\":[],\"httpDownloadTactics\":[],\"tracerouteTactics\":[],\"httpUploadTactics\":[],\"dnsTactics\":[]},\"startTime\":1503458160,\"endTime\":1506136560,\"cycleTime\":10,\"collectTime\":5,\"collectType\":0,\"kafkaUrl\":\"10.0.0.210:9092\"}],\"status\":\"1\"}";
    return cJSON_Parse(buf);
}
#endif

int platform_init(Platform_t* platform, char* svr_host, uint16_t tcp_port, uint16_t udp_port, char* scheme)
{
    // load_probe_version(PROBE_VERSION_FILE, commReqParam.version, sizeof(commReqParam.version));
    strcpy(platform->addr.host, svr_host);
    strcpy(platform->addr.scheme, scheme);
    platform->addr.tcp_port = tcp_port;
    platform->addr.udp_port = udp_port;
    TAILQ_INIT(&platform->resp.task.head);

    // platform->tpool = thread_pool_new(1);
    // thread_pool_set_name(platform->tpool, "platform task");
    // thread_pool_set_auto_extern(platform->tpool);

    // platform->resp.task.tpool = thread_pool_new(1);
    // thread_pool_set_name(platform->resp.task.tpool, "Task");
    // thread_pool_set_auto_extern(platform->resp.task.tpool);
    return 1;
}
static cJSON *get_auth_json_from_file(){
    FILE *fp;
    cJSON *resp = NULL;
    char buf[1024] = {0};

    fp = fopen(AUTH_JSON_FILE, "r");
    if(!fp){
        log_syserr("get_auth_json_from_file failed.");
        return NULL;
    }

    if(fgets(buf, sizeof(buf), fp) == NULL){
        log_syserr("fgets failed");
        fclose(fp);
        return NULL;
    }
    log_trace("auth json:%s", buf);
    fclose(fp);

    resp = cJSON_Parse(buf);
    log_trace("resp:%p", resp);
    return resp;
}
static void save_auth_json_to_file(cJSON *resp){
    FILE *fp = NULL;
    char *ptr = NULL;

    fp = fopen(AUTH_JSON_FILE, "w");
    if(!fp){
        log_syserr("save_auth_json_to_file error.");
        return;
    }

    ptr = cJSON_PrintUnformatted(resp);
    if(!ptr){
        log_trace("cJSON_PrintUnformatted ptr is NULL.");
        fclose(fp);
        return;
    }
    fputs(ptr, fp);
    
    fclose(fp);
    free(ptr);
}
int platform_auth(Platform_t* platform)
{
    char url[1024] = { 0 };
    char* req = NULL;
    int ret = 0, auth_from_platform_flag = 0;
    cJSON *resp = NULL, *root = NULL, *json_ptr = NULL;

    if((resp = get_auth_json_from_file()) == NULL){

        auth_from_platform_flag = 1;

        log_trace("tcp port:%d", platform->addr.tcp_port);
        http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port,
            WEB_AUTH_PATH, url);

        root = crq2cjson(&commReqParam, root);
        root = devinfo2cjson(&devInfo, root);
        req = cJSON_Print(root);
        cJSON_Delete(root);

    #if _TEST_DATA_
        resp = __Test_dial_auth_data();
    #else
        if (!(resp = http_post(url, req))) {
            log_error("Auth:http post %s error.", url);
            goto out;
        }
    #endif

        if (!cJSON_IsObject(resp)) {
            log_error("Auth:JSON Parse failed.");
            goto out;
        }

        if (!check_platform_resp(resp)) {
            goto out;
        }

        save_auth_json_to_file(resp);
    }
    
    json_ptr = cJSON_GetObjectItem(resp, "token");
    if (json_ptr)
        strcpy(platform->resp.token, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "probeId");
    if (json_ptr)
        strcpy(commReqParam.probeId, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "cityName");
    if (json_ptr)
        strcpy(commReqParam.cityName, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "cityAlias");
    if (json_ptr)
        strcpy(commReqParam.cityAlias, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "cityId");
    if (json_ptr)
        strcpy(commReqParam.cityId, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "province");
    if (json_ptr)
        strcpy(commReqParam.province, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "bdaccount");
    if (json_ptr && !cJSON_IsNull(json_ptr))
        strcpy(commReqParam.bdaccount, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "password");
    if (json_ptr && !cJSON_IsNull(json_ptr))
        strcpy(commReqParam.bdpassword, json_ptr->valuestring);

    json_ptr = cJSON_GetObjectItem(resp, "bdcode");
    if (json_ptr) {
        strcpy(commReqParam.bdcode, json_ptr->valuestring);
        commReqParam.bdcode_val = atoi(commReqParam.bdcode) * 1000 * 1000;
    }

    json_ptr = cJSON_GetObjectItem(resp, "logReportUrl");
    if(json_ptr){
        platform->resp.logReportUrl = strdup(json_ptr->valuestring);
    }

    json_ptr = cJSON_GetObjectItem(resp, "clientPerformanUrl");
    if(json_ptr){
        platform->resp.clientPerformanUrl = strdup(json_ptr->valuestring);
    }

    //如果认证信息是从平台获取的而不是本地获取的，则可以更新系统时间
    if(auth_from_platform_flag == 1){
        json_ptr = cJSON_GetObjectItem(resp, "systemTime");
        if (json_ptr) {
            if (devInfo.needUpdateSysTime){
                devinfo_update_systime(json_ptr->valueint);
            }
        }
    }
    

    ret = 1;

// dev_use_user_pwd_redial(commReqParam.bdaccount, commReqParam.bdpassword);
out:
    if(req)
        free(req);
    if (resp)
        cJSON_Delete(resp);
    return ret;
}
static cJSON* _dial_req_gen_data(Platform_t* platform, cJSON* root)
{
    int i = 0;
    struct Dial_Task_s* item;
    cJSON *obj, *_root;

    if (!root)
        _root = cJSON_CreateObject();
    else
        _root = root;

    cJSON* tasklist = cJSON_CreateArray();

    cJSON_AddStringToObject(_root, "token", platform->resp.token);

    TAILQ_FOREACH(item, &platform->resp.task.head, entry)
    {
        obj = cJSON_CreateObject();
        cJSON_AddStringToObject(obj, "dialTaskId", item->id);
        cJSON_AddStringToObject(obj, "version", item->version);
        cJSON_AddItemToArray(tasklist, obj);
    }
    cJSON_AddItemToObject(_root, "dialtaskList", tasklist);
    return _root;
}
static void _save_dial_task_strategy(cJSON* cStrategy){
    #define TMP_STRATEGY_FILE "strategy.json"
    FILE *fp;
    char *ptr = NULL;

    fp = fopen(TMP_STRATEGY_FILE, "w");
    if(!fp){
        return;
    }

    ptr = cJSON_Print(cStrategy);
    cJSON_Minify(ptr);
    fprintf(fp, "%s", ptr);

    free(ptr);
    fclose(fp);
}
static void _save_task_info(Dial_Task_t* dial_task){
    
    FILE *fp = NULL;

    fp = fopen(DIAL_TASK_SAVE_ID_FILE, "a");
    if(fp){
        fprintf(fp, "id:%s, startTime:%d, endTime:%d, cycleTime:%d, collectTime:%d\n", 
            dial_task->id, dial_task->startTime, dial_task->endTime, dial_task->cycleTime, dial_task->collectTime);
        fclose(fp);
    }
}
int platform_dial_req(Platform_t* platform)
{
    char url[1024] = { 0 };
    char* req;
    int arr_len, i = 0, ret = 0;
    cJSON *resp, *root = NULL, *tasklist = NULL, *json_ptr = NULL, *json_obj = NULL;
    Dial_Task_t* dial_task;

    //删除拨测任务ID记录文件
    remove(DIAL_TASK_SAVE_ID_FILE);

    http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port,
        WEB_DIAL_REQ_PATH, url);

    root = crq2cjson(&commReqParam, root);
    cJSON_AddStringToObject(root, "bcdcode", commReqParam.bdcode);

    root = _dial_req_gen_data(platform, root);
    req = cJSON_Print(root);
    cJSON_Delete(root);

#if _TEST_DATA_
    resp = __Test_dial_req_data();
#else
    if (!(resp = http_post(url, req))) {
        log_error("Dial Task Request:http post %s error.", url);
        goto out;
    }
#endif
    if (!cJSON_IsObject(resp)) {
        log_error("Dial Task Request:response is not Object.");
        goto out;
    }
    if (!check_platform_resp(resp)) {
        goto out;
    }

    tasklist = cJSON_GetObjectItem(resp, "dialTaskList");
    if (!cJSON_IsArray(tasklist)) {
        log_error("Dial Task Request:Dial Task is not Array.");
        goto out;
    }

    // _save_dial_task_strategy(resp);

    arr_len = cJSON_GetArraySize(tasklist);
    platform->resp.task.count = arr_len;

    for (i = 0; i < arr_len; i++) {
        dial_task = calloc(1, sizeof(Dial_Task_t));

        json_obj = cJSON_GetArrayItem(tasklist, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "dialTaskId");
        strcpy(dial_task->id, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "version");
        strcpy(dial_task->version, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "startTime");
        dial_task->startTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "endTime");
        dial_task->endTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "cycleTime");
        dial_task->cycleTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "collectTime");
        dial_task->collectTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaUrl");
        dial_task->kafka_url = strdup(json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        strcpy(dial_task->kafka_topic, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "isdefault");
        if (json_ptr)
            strcpy(dial_task->isdefault, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "dialTactics");
        dial_tactics_parse(&dial_task->tactics, json_ptr);

        TAILQ_INSERT_TAIL(&platform->resp.task.head, dial_task, entry);

        _save_task_info(dial_task);
    }
    ret = 1;
out:
    free(req);
    if (resp)
        cJSON_Delete(resp);
    return ret;
}
#if 0
static void* _each_task_ping_report(void* arg)
{
    void** arg_ptr = arg;
    Platform_t* platform = (Platform_t*)arg_ptr[0];
    ping_tactics_t* ptr = (ping_tactics_t*)arg_ptr[1];
    free(arg);

    char* post_data = NULL, reportTime[32], collectTime[32];
    cJSON *data;

    while (1) {
        pthread_mutex_lock(&ptr->mutex);
        pthread_cond_wait(&ptr->cond, &ptr->mutex);

        data = cJSON_CreateObject();
        cJSON_AddStringToObject(data, "dialTaskId", ptr->taskId);
        cJSON_AddStringToObject(data, "version", ptr->version);
        cJSON_AddStringToObject(data, "indexType", "ping");
        get_timestring(reportTime);
        cJSON_AddStringToObject(data, "reportTime", reportTime);
        get_timestring(collectTime);
        cJSON_AddStringToObject(data, "collectTime", collectTime);
        cJSON_AddItemToObject(data, "pingList", ptr->result);
        post_data = cJSON_Print(data);
        log_debug("post_data:%s", post_data);
        cJSON_Delete(ptr->result);
        free(post_data);
        pthread_mutex_unlock(&ptr->mutex);
    }
    return NULL;
}
static void* _each_task_report(void* arg)
{
    void** arg_ptr = arg;
    cJSON* report_data;
    Platform_t* platform = (Platform_t*)arg_ptr[0];
    struct Dial_Task_s* ptr = (struct Dial_Task_s*)arg_ptr[1];
    free(arg);

    thread_pool_t* tpool = thread_pool_new(1);
    thread_pool_set_auto_extern(tpool);
    thread_pool_set_name(tpool, "task report");

    arg_ptr = calloc(2, sizeof(void*));
    arg_ptr[0] = (void*)platform;
    arg_ptr[1] = (void*)&ptr->tactics.ping;
    strcpy(ptr->tactics.ping.taskId, ptr->id);
    strcpy(ptr->tactics.ping.version, ptr->version);
    thread_pool_add_task(tpool, _each_task_ping_report, (void*)arg_ptr);

    thread_pool_wait(tpool);
    thread_pool_destroy(tpool);

    return NULL;
}
void* platform_report(void* arg)
{
    thread_pool_t* tpool = thread_pool_new(1);
    thread_pool_set_auto_extern(tpool);
    thread_pool_set_name(tpool, "platform report");

    Platform_t* platform = (Platform_t*)arg;
    struct Dial_Task_Head* head = &platform->resp.task_head;
    struct Dial_Task_s* ptr;
    void** thread_arg;
    TAILQ_FOREACH(ptr, head, entry)
    {
        thread_arg = calloc(2, sizeof(void*));
        thread_arg[0] = (void*)platform;
        thread_arg[1] = (void*)ptr;
        thread_pool_add_task(tpool, _each_task_report, (void*)thread_arg);
    }
    thread_pool_wait(tpool);
    thread_pool_destroy(tpool);

    return NULL;
}
#endif
int platform_push_probe_upload_report(Platform_t* platform, int cmd_id, int code)
{
    cJSON *root = NULL, *resp;
    char* comm_req = NULL;
    char url[512];

    commReqParam.cmdId = cmd_id;
    commReqParam.code = code;
    root = crq2cjson(&commReqParam, root);
    comm_req = cJSON_Print(root);
    cJSON_Delete(root);
    http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port, WEB_PUSH_PROBE_UPLOAD_PATH, url);
    if ((resp = http_post(url, comm_req)) == NULL) {
        log_error("HTTP Push probe upload failed.");
        goto out;
    }
    if (resp)
        cJSON_Delete(resp);
out:
    commReqParam.cmdId = 0;
    commReqParam.status = 0;
    commReqParam.code = 0;
    memset(commReqParam.msg, 0, sizeof(commReqParam.msg));
    free(comm_req);
    return 1;
}
int platform_download_probe(Platform_t* platform)
{
    char version[32] = { 0 };
    char url[512] = {0}, *buffer = NULL, *req = NULL, down_url[512];
    int len = 0;
    FILE* fp = NULL;
    cJSON *root = NULL, *resp = NULL;
    int ret = 0;

    root = crq2cjson(&commReqParam, root);
    cJSON_AddStringToObject(root, "clientFactory", devInfo.factory);
    req = cJSON_Print(root);
    cJSON_Delete(root);

    http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port, WEB_DOWNLOAD_PROBE_PATH, url);
    if (http_download_get_url(url, req, down_url, version) == 0) {
        log_error("HTTP download get down url failed");
        goto out;
    }

    if (strcmp(version, commReqParam.version) == 0) {
        log_info("Local Probe version is same as remote version, no need download.");
        ret = 2;
        goto out;
    }

    save_probe_version(PROBE_VERSION_FILE, version, strlen(version));
    load_probe_version(PROBE_VERSION_FILE, commReqParam.version, sizeof(commReqParam.version));

    if (http_download_file(down_url, PROBE_TAR_FILE) == 0) {
        log_error("HTTP download Probe failed.");
        goto out;
    }

    if (0 != system("tar xzf " PROBE_TAR_FILE)) {
        if (0 != system("busybox tar xzf " PROBE_TAR_FILE)) {
            log_syserr("Untar " PROBE_TAR_FILE " failed.");

            log_trace("Move tar file to " ONU_PROBE_EXE);
            system("mv " PROBE_TAR_FILE " " ONU_PROBE_EXE);
            system("chmod a+x " ONU_PROBE_EXE);

            goto out;
        }
    }
    ret = 1;
out:
    free(req);
    return ret;
}
static void _set_nodeurl(Platform_t* platform)
{
    struct Dial_Task_s* ptr = NULL;
    TAILQ_FOREACH(ptr, &platform->resp.task.head, entry)
    {
        http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port,
            WEB_SPEED_NODE_PATH, ptr->nodeUrl);
        strcpy(ptr->probeId, commReqParam.probeId);
    }
}
static void _kill_all(){
    system_killall(ONU_PROBE_EXE);
}
static void _restart_exe(Platform_t* platform){
    char cmd[256] = {0};

    sprintf(cmd, "./%s -a %s -n %d -u %d &", 
            ONU_PROBE_EXE,
            platform->addr.host, 
            platform->addr.tcp_port,
            platform->addr.udp_port);
    log_trace("%s", cmd);
    system(cmd);
}
static void kill_self(Platform_t* platform, void (*restart_cb)(Platform_t* platform)){
    int pid = 0, tmp_pid = 0;
    char exec_name[32] = {0}, tmp_name[32] = {0};
    DIR *dir = NULL;
    struct dirent *dptr = NULL;

    pid = getpid();
    get_exec_name_by_pid(pid, exec_name);
    if(strlen(exec_name) == 0){
        log_error("get exec name failed :%s", exec_name);
        _kill_all();
        return;
    }

    dir = opendir("/proc");
    if(!dir){
        log_syserr("open /proc dir failed.");
        _kill_all();
        return;
    }

    while((dptr = readdir(dir)) != NULL){
        if(strcmp(dptr->d_name, ".") == 0 || strcmp(dptr->d_name, "..") == 0){
            continue;
        }
        if((tmp_pid = atoi(dptr->d_name)) > 0){
            bzero(tmp_name, sizeof(tmp_name));
            get_exec_name_by_pid(tmp_pid, tmp_name);
            if(!strcmp(tmp_name, exec_name) && pid != tmp_pid){
                log_notice("kill %u", tmp_pid);
                kill(tmp_pid, SIGKILL);
            }
        }
    }
    closedir(dir);

    if(restart_cb){
        restart_cb(platform);
    }

    kill(pid, SIGKILL);//kill self process
}

static int _deal_cmd_id(Platform_t* platform, uint32_t cmd_id)
{
    int flag = 0, ret = 0;
    pthread_t thid;
    int dialtask_all_count = 0;

    log_notice("Deal cmd id:%d", cmd_id);
    switch (cmd_id) {
    case 1: //pushProbe
        log_notice("Start to download "PROBE_TAR_FILE);
        ret = platform_download_probe(platform);
        if (ret == 0) {
            log_info("Probe Update failed.");
            platform_push_probe_upload_report(platform, cmd_id, 0);
        } else {//若与服务器是相同版本，则重启onuProbe
            platform_push_probe_upload_report(platform, cmd_id, 1);
            log_info("Probe Update successfully, Restart...");
            remove(AUTH_JSON_FILE);
            return 1;
        }
        break;

    case 21:
    case 2: //pushDialTactics
        //当有任务更新时，先设置当前所有任务取消，待所有任务退出后，重新获取所有任务。
        dial_task_all_set_flag(&platform->resp.task, DIAL_TASK_FLAG_CANCEL);
        /* 当收到命令2，判断当前是否有任务，若有任务，则等所有任务退出并待收到21命令后再更新策略；若无任务，直接更新策略
         当收到命令21， 则表明所有任务退出，直接更新策略 */
        dialtask_all_count = dial_task_get_running_count(&platform->resp.task);
        log_trace("dialtask_all_count:%d, cmd_id:%d", dialtask_all_count, cmd_id);
        if ((dialtask_all_count == 0 && cmd_id == 2) || cmd_id == 21) {
            log_notice("Update dial task list.");


            //当任务都退出后，重新获取任务， 21 为任务退出时，线程发送的通知命令，不是平台下发的命令。
            dial_task_free_all(&platform->resp.task);
            if (!platform_auth(platform)) {
                log_warn("Platform Auth failed.");
                platform_push_probe_upload_report(platform, cmd_id, 0);
                kill_self(platform, NULL);
            }

            if (!platform_dial_req(platform)) {
                log_warn("Platform request dial task failed.");
                platform_push_probe_upload_report(platform, cmd_id, 0);
                kill_self(platform, NULL);
            }
            _set_nodeurl(platform);
            // thread_pool_add_task(platform->tpool, dial_task_run, (void*)&platform->resp.task.head);
            pthread_create(&thid, NULL, dial_task_run, (void*)&platform->resp.task.head);
            pthread_detach(thid);
        }
        break;
    case 31: //probeCtrl
    case 32:
        if (cmd_id == 31)
            flag = DIAL_TASK_FLAG_RUN;
        if (cmd_id == 32)
            flag = DIAL_TASK_FLAG_STOP;
        dial_task_all_set_flag(&platform->resp.task, flag);
        platform_push_probe_upload_report(platform, cmd_id, 1);
        break;
    case 4: //主动测试
        pthread_create(&thid, NULL, specify_dial_task, (void*)platform);
        pthread_detach(thid);
        break;
    case 5://上报日志
        log_upload_report(platform->resp.logReportUrl, commReqParam.probeId, devInfo.mac);
        break;
    default:
        log_error("Unkown cmd id:%d", cmd_id);
        break;
    }
    return 0;
}
static int _check_if_need_update(Platform_t *platform){

    #define _CHECK_UPDATE_INTERVAL 86400

    time_t now;

    now = time(NULL);

    if(_Check_update_tm == 0){
        _Check_update_tm = now + ((int)(get_random()/(RAND_MAX + 1.0) * _CHECK_UPDATE_INTERVAL));
        log_trace("_Check_update_tm:%d", _Check_update_tm);
        return 0;
    }

    if(now >= _Check_update_tm){
        if(platform_download_probe(platform) == 1){// 1: new version
            return 1;
        }else{
            _Check_update_tm = now + _CHECK_UPDATE_INTERVAL;
            return 0;
        }
    }

    return 0;
}
static void* platform_check_lastcmd(void* arg)
{

    Platform_t* platform = (Platform_t*)arg;
    struct sockaddr_in addr, peeraddr;
    int addr_len = 0;
    int fd, cli_fd;
    int ret;
    uint32_t cmd_id;
    int reuse_addr_on = 1;
    fd_set rdfds;
    struct timeval tmo;

    addr_len = sizeof(peeraddr);


    log_info("Check LastCmd Run... (pid:%u)", getpid());
retry:
    fd = socket(AF_INET, SOCK_STREAM, 0);
    if(fd < 0){
        log_syserr("platform_check_lastcmd create socket failed.");
        sleep(1);
        goto retry;
    }
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse_addr_on, sizeof(reuse_addr_on));
    
    sockaddr_set(&addr, AF_INET, htons(HEART_LISTNE_PORT), "127.0.0.1");

    if (bind(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        log_syserr("Platform Check LastCmd bind error.");
        close(fd);
        sleep(1);
        goto retry;
    }
    if (listen(fd, 5) < 0) {
        log_syserr("Platform Check LastCmd listen error.");
        close(fd);
        sleep(1);
        goto retry;
    }
    while (1) {
        FD_ZERO(&rdfds);
        FD_SET(fd, &rdfds);
        tmo.tv_sec = 10;
        tmo.tv_usec = 0;
        ret = select(fd + 1, &rdfds, NULL, NULL, &tmo);
        if(ret == 0){
            if(_check_if_need_update(platform)){
                remove(AUTH_JSON_FILE);
                log_info("need update self ...");
                close(fd);
                kill_self(platform, _restart_exe);
            }
            continue;
        }else if(ret < 0){
            log_syserr("select error.");
            close(fd);
            goto retry;
        }

        if(FD_ISSET(fd, &rdfds)){
            cmd_id = 0;
            log_trace("fd:%d", fd);
            cli_fd = accept(fd, (struct sockaddr *)&peeraddr, &addr_len);
            log_trace("");
            if (cli_fd > 0) {
                if(0 > read(cli_fd, &cmd_id, sizeof(cmd_id))){
                    log_syserr("platform_check_lastcmd read error.");
                    close(cli_fd);
                    continue;
                }
                close(cli_fd);
                ret = _deal_cmd_id(platform, cmd_id);
                if(ret == 1){
                    log_info("kill self and restart.");
                    close(fd);
                    kill_self(platform, _restart_exe);
                }
            }else{
                if(cli_fd < 0){
                    log_syserr("platform_check_lastcmd accept error.");
                    close(fd);
                    sleep(1);
                    goto retry;
                }
            }
        }
        
    }
}
static char * _get_process_stat_data(process_info_t *pinfo, int len){
    int i = 0;
    char tmpstr[32] = {0};
    char reportTime[32] = {0};
    char *data = NULL;
    int thread_count = 0;
    cJSON *root = NULL, *arr = NULL, *obj = NULL;

    root = cJSON_CreateObject();
    cJSON_AddStringToObject(root, "mac", devInfo.mac);

    get_timestring(reportTime);
    cJSON_AddStringToObject(root, "reportTime", reportTime);

    arr = cJSON_CreateArray();
    for(i =0; i < len; i++){
        obj = cJSON_CreateObject();
        cJSON_AddStringToObject(obj, "collectTime", pinfo[i].timestamp);

        sprintf(tmpstr, "%.2f", pinfo[i].cpu_usage);
        cJSON_AddStringToObject(obj, "cpuUsage", tmpstr);

        sprintf(tmpstr, "%.2f", pinfo[i].process_cpu_usage);
        cJSON_AddStringToObject(obj, "probeCpuUsage", tmpstr);

        sprintf(tmpstr, "%.2f", pinfo[i].mem_usage);
        cJSON_AddStringToObject(obj, "memUsage", tmpstr);

        sprintf(tmpstr, "%.2f", pinfo[i].process_mem_usage);
        cJSON_AddStringToObject(obj, "probeMemUsage", tmpstr);

        cJSON_AddNumberToObject(obj, "freeMem", pinfo[i].mem_free);
        if(pinfo[i].thread_count == 1){
            thread_count = pinfo[i].sub_process_count;
        }else{
            thread_count = pinfo[i].thread_count;
        }
        cJSON_AddNumberToObject(obj, "thread", thread_count);
        cJSON_AddNumberToObject(obj, "fd", pinfo[i].fd_count);
        cJSON_AddNumberToObject(obj, "socket", pinfo[i].socket_count);
        
        cJSON_AddItemToArray(arr, obj);
    }

    cJSON_AddItemToObject(root, "stateList", arr);

    data = cJSON_Print(root);

    cJSON_Delete(root);

    return data;
}
static int check_process_info(process_info_t *pinfo){
    if(pinfo->fd_count > 100 
        || pinfo->sub_process_count > 50 
        || pinfo->thread_count > 50 
        || pinfo->process_mem_usage > 10)
    {
        //超过某值开始显示信息
        process_self_print(pinfo);

        //超过最大值则退出程序
        if(pinfo->fd_count > 900 
        || pinfo->sub_process_count > 100 
        || pinfo->thread_count > 100 
        || pinfo->process_mem_usage > 20)
        {
            return -1;
        }
    }

    

    return 1;    
}
void *process_stat_monitor(void *arg){

    int i = 0;
    int collectInterval = 15, reportInterval = 15, collectSize = 0;
    int tmp_collectInterval = 0, tmp_reportInterval = 0;
    cJSON * resp = NULL, *cjson_collect = NULL, *cjson_report = NULL;
    char *data = NULL;
    char collectTime[32] = {0}, reportTime[32] = {0};
    Platform_t *platform = NULL;
    process_info_t *pinfo = NULL;

    log_info("Process Monitor Run... (pid:%u)", getpid());

    platform = (Platform_t *)arg;
    
    collectSize = reportInterval / collectInterval;

    pinfo = calloc(collectSize, sizeof(process_info_t));
    if(!pinfo){
        log_syserr("alloc process info struct failed.");
    }

    while(1){
        sleep(collectInterval);
        process_info(&pinfo[i]);
        if(check_process_info(&pinfo[i]) < 0){
            log_error("Process use more resource, exit.");
            kill_self(platform, _restart_exe);
        }
        // log_info("process monitor i:%d", i);
        if(Debug_mode()){
            process_self_print(&pinfo[i]);
        }
        // process_info_print(&pinfo[i]);
        i++;
        if(i >= collectSize){
            data = _get_process_stat_data(pinfo, collectSize);
            if(data){
                resp = http_post(platform->resp.clientPerformanUrl, data);
                if(resp){
                    if(cJSON_IsObject(resp)){
                        cjson_collect = cJSON_GetObjectItem(resp, "collectTime");
                        if(cjson_collect){
                            tmp_collectInterval = atoi(cjson_collect->valuestring);
                        }
                        cjson_report = cJSON_GetObjectItem(resp, "reportTime");
                        if(cjson_report){
                            tmp_reportInterval = atoi(cjson_report->valuestring);
                        }
                        if(tmp_collectInterval != collectInterval || tmp_reportInterval != reportInterval){
                            free(pinfo);
                            reportInterval = tmp_reportInterval;
                            collectInterval = tmp_collectInterval;
                            collectSize = reportInterval / collectInterval;
                            pinfo = calloc(collectSize, sizeof(process_info_t));
                            if(!pinfo){
                                log_syserr("realloc process info struct failed.");
                            }
                        }
                    }
                    cJSON_Delete(resp);
                }
                free(data);
            }
            memset(pinfo, 0, sizeof(process_info_t) * collectSize);
            i = 0;
        }
    }
}
int platform_task_run(Platform_t* platform)
{
    // thread_pool_t* tpool = platform->tpool;
    pthread_t tid;
    _set_nodeurl(platform);


    pthread_create(&tid, NULL, heart_bit_loop, (void*)&platform->heartinfo);
    pthread_detach(tid);
    pthread_create(&tid, NULL, process_stat_monitor, (void*)platform);
    pthread_detach(tid);
    pthread_create(&tid, NULL, dial_task_run, (void*)&platform->resp.task.head);
    pthread_detach(tid);

    platform_check_lastcmd((void*)platform);

    return 1;
}